package com.jsy.mqhelper.charlotterpc.register.mq.rabbit;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.concurrent.TimeoutException;

import com.jsy.mqhelper.charlotterpc.core.mq.rabbit.RabbitConnectionFactory;
import com.jsy.mqhelper.charlotterpc.protocol.CharlotteRpcRequest;
import com.jsy.mqhelper.charlotterpc.protocol.CharlotteRpcResponse;
import com.jsy.mqhelper.charlotterpc.protocol.ResponseCode;
import com.jsy.mqhelper.charlotterpc.register.RemoteRegisterHandler;
import com.jsy.mqhelper.charlotterpc.util.CharlotteRpcUtil;
import com.jsy.mqhelper.common.core.MetaFunction;
import com.jsy.mqhelper.common.core.MetaInterface;
import com.jsy.mqhelper.common.core.Registry;
import com.jsy.mqhelper.common.exception.MetaFunctionNotFoundException;
import com.jsy.mqhelper.common.util.JsonUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import lombok.extern.slf4j.Slf4j;

import static com.jsy.mqhelper.common.constant.Constants.MQ_HELPER;

/**
 * @author: SongyangJi
 * @description:
 * @since: 2021/9/26
 */
@Slf4j(topic = MQ_HELPER)
public class RabbitMQRegisterHandler extends RemoteRegisterHandler {

    RabbitConnectionFactory rabbitConnectionFactory;

    public RabbitMQRegisterHandler(Registry localRegistry, RabbitConnectionFactory rabbitConnectionFactory) {
        super(localRegistry);
        this.rabbitConnectionFactory = rabbitConnectionFactory;
    }

    @Override
    protected void register(Registry registry) {
        log.info("register rpc services in rabbitmq");
        log.info("rpc interfaces={}", registry.getAllInterfaceNames());
        try {
            doRegister(registry.getAllMetaInterfaces());
        } catch (IOException | TimeoutException e) {
            log.error("register fail", e);
        }
    }

    private void doRegister(Collection<MetaInterface> interfaces) throws IOException, TimeoutException {
        Connection connection = rabbitConnectionFactory.createConnection();
        for (MetaInterface metaInterface : interfaces) {
            /* one service(interface) one channel */
            Channel channel = connection.createChannel();
            String rpcQueueName = CharlotteRpcUtil.generateQueueName(metaInterface);
            channel.queueDeclare(rpcQueueName, true, false, false, null);
            // TODO just to debug
            channel.queuePurge(rpcQueueName);

            /* channel shutdown monitor */
            channel.addShutdownListener(new ShutdownListener() {
                @Override
                public void shutdownCompleted(ShutdownSignalException cause) {
                    log.info("isInitiatedByApplication={}", cause.isInitiatedByApplication());
                    log.info("isHardError={}", cause.isHardError());
                    log.info("getReference={}", cause.getReference().getClass());
                    log.warn("shutdownCompleted", cause);
                }
            });

            DeliverCallback deliverCallback = (consumerTag, message) -> {
                log.debug("'consumerTag={}, message.envelope={}, message.body={}, message.properties={}", consumerTag,
                    message.getEnvelope(), new String(message.getBody()), message.getProperties());
                String rpcRequestStr = new String(message.getBody(), StandardCharsets.UTF_8);
                CharlotteRpcRequest rpcRequest = JsonUtil.parse(rpcRequestStr, CharlotteRpcRequest.class);
                log.debug("charlotteRpcRequest={}", rpcRequest);

                String methodId = rpcRequest.getMethodId();
                Object[] params = rpcRequest.getParams();
                if (params != null && params.length > 0) {
                    log.debug("params are:");
                    for (Object param : params) {
                        log.debug(" - param={}, paramClass={}", param, param.getClass());
                    }
                }

                /* 正常返回的结果 */
                Object result = null;
                /* 正常执行抛出的异常 */
                Throwable exception = null;
                /* 内部错误 */
                CharlotteRpcResponse.Error error = null;
                try {
                    MetaFunction metaFunction = this.localRegistry.access(methodId);
                    Method method = metaFunction.getMethod();
                    Object handler = metaFunction.getHandler();
                    result = method.invoke(handler, params);
                } catch (MetaFunctionNotFoundException metaFunctionNotFoundException) {
                    error = new CharlotteRpcResponse.Error(ResponseCode.METHOD_NOT_FOUND,
                        metaFunctionNotFoundException.getMessage());
                    log.error("methodId={}, error={}", methodId, error);
                    // todo debug
                    metaFunctionNotFoundException.printStackTrace();
                } catch (IllegalAccessException illegalAccessException) {
                    error = new CharlotteRpcResponse.Error(ResponseCode.OTHER, illegalAccessException.getMessage());
                    log.error("methodId={}, error={}", methodId, error);
                    // todo debug
                    illegalAccessException.printStackTrace();
                } catch (InvocationTargetException invocationTargetException) {
                    exception = invocationTargetException.getTargetException();
                    log.warn("methodId={}, exception={}", methodId, exception);
                    // todo debug
                    invocationTargetException.printStackTrace();
                } finally {
                    CharlotteRpcResponse rpcResponse;
                    if (error == null && exception == null) { // 方法调用一切正常
                        rpcResponse = new CharlotteRpcResponse(result);
                    } else if (error != null) {
                        rpcResponse = new CharlotteRpcResponse(error);
                    } else {
                        rpcResponse = new CharlotteRpcResponse(exception);
                    }
                    // channel 同时收发是线程安全的
                    channel.basicPublish("", message.getProperties().getReplyTo(), message.getProperties(),
                        JsonUtil.stringify(rpcResponse).getBytes(StandardCharsets.UTF_8));
                    channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
                }

            };
            channel.basicConsume(rpcQueueName, false, deliverCallback, consumerTag -> {
            });
        }
    }

}